/* Definitions for the stream type.

   This file is part of khipu.

   khipu is free software: you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published by
   the Free Software Foundation; either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public License
   along with this program.  If not, see <https://www.gnu.org/licenses/>.  */

#include <cstdlib>
#include <cstdio>
#include <climits>
#include "stream.hpp"
#include "array.hpp"
#include "bvector.hpp"
#include "str.hpp"
#include "memory.hpp"
#include "thread.hpp"
#include "integer.hpp"
#include "symbol.hpp"
#include "llio.hpp"

KP_DECLS_BEGIN

static void
fini_stream (finobj *self)
{
  ((stream *)self)->close (interpreter::self ());
}

spos spos::decode (object obj)
{
  spos ret (0);
  object head;

  if (array_p (obj))
    {
      head = xaref (obj, 0);
      ret.state = xaref (obj, 1);
    }
  else
    head = obj;

  if (fixint_p (head))
    ret.offset = as_int (head);
  else
    {
      int64_t off;
      bigint_as (head, off);
      ret.offset = off;
    }

  return (ret);
}

object spos::encode (interpreter *interp)
{
  object ret = intobj (interp, this->offset);

  if (this->state != UNBOUND)
    {
      valref tmp (interp, ret);
      ret = alloc_array (interp, 2);
      xaref(ret, 0) = *tmp, xaref(ret, 1) = this->state;
    }

  kp_return (ret);
}

bool stream::xseek (interpreter *interp, const spos& npos, int whence)
{
  spos tmp = npos;
  bool ret = this->ops->seek (interp, *this, tmp, whence);
  if (ret)
    this->pos = tmp;

  return (ret);
}

stream* stream::make (interpreter *interp, int mode,
  unsigned int bufsiz, const xops *ops, void *cookie)
{
  evh_guard eg (interp);
  stream *ret = alloch<stream> ();
  ret->ops = ops;

  ret->cookie = cookie;
  ret->bvec = alloc_bvector (interp,
    bufsiz << ((mode & STRM_BID) / STRM_BID));

  ret->rdbuf.init ((char *)as_bvector(ret->bvec)->data, bufsiz);
  ret->rdbuf.bump (bufsiz);

  char *wrbuf = ret->rdbuf.start + ((mode & STRM_BID) ? bufsiz : 0);
  ret->wrbuf.init (wrbuf, bufsiz);

  valref bv (interp, ret->bvec);
  ret->ilock = (mode & STRM_NOLOCK) ? UNBOUND : alloc_lock (interp, true);
  ret->io_flags = mode;

  if (!(ret->io_flags & STRM_APP))
    ret->pos = spos (0);
  else
    ret->io_flags &= ~STRM_OFFSET;

  ret->extra = UNBOUND;
  ret->fini = fini_stream;
  ret->vo_full |= FLAGS_FINALIZABLE;
  interp->alval = ret->as_obj ();
  gc_register (interp, ret);
  return (ret);
}

int stream::bufsize () const
{
  return (as_bvector(this->bvec)->nbytes);
}

bool stream::write_full (interpreter *interp,
  const void *src, uint32_t bytes)
{
  while (bytes > 0)
    {
      int tmp = this->xwrite (interp, src, bytes);
      if (tmp <= 0)
        {
          this->io_flags |= STRM_ERR;
          return (false);
        }
      
      src = (const char *)src + tmp, bytes -= tmp;
    }

  return (true);
}

bool stream::rflush (interpreter *interp)
{
  if (this->io_flags & STRM_OFFSET)
    // The current offset is always valid - Just seek from the beginning.
    return (this->xseek (interp, this->pos, SEEK_SET));
  
  // Get the current offset and seek from the beginning.
  spos npos (0);
  if (!this->xseek (interp, npos, SEEK_CUR) ||
      !this->xseek (interp, npos, SEEK_SET))
    return (false);
  
  this->pos = npos;
  this->reset_buff ();
  return (true);
}

bool stream::flush (interpreter *interp)
{
  return ((this->io_flags & STRM_LASTW) ? this->wflush (interp) : true);
}

bool stream::refill (interpreter *interp)
{
  if (this->io_flags & STRM_EOS)
    return (false);

  int nr = this->xread (interp, this->rdbuf.start, this->bufsize ());

  if (nr <= 0)
    {
      this->io_flags |= nr == 0 ? STRM_EOS : STRM_ERR;
      this->rdbuf.set (this->rdbuf.end);
      return (false);
    }

  this->rdbuf.set (this->rdbuf.start);
  this->rdbuf.end = this->rdbuf.start + nr;
  return (true);
}

int stream::read (interpreter *interp, void *dst, uint32_t bytes)
{
  int cnt, nbytes = (int)bytes;
  char *base = (char *)dst;

  if (this->need_wflush () && !this->wflush (interp))
    return (-1);

  this->set_lastr ();
  /* We need to loop because the buffer may not be fully
   * filled on each iteration... */
  while (nbytes > (cnt = this->rdbuf.left ()))
    {
      memcpy (dst, this->rdbuf.curr, cnt);
      dst = (char *)dst + cnt, nbytes -= cnt;
      if (!this->refill (interp))
        { // Partial result.
          this->add_pos ((char *)dst - base);
          return ((char *)dst - base);
        }
    }

  if (this->io_flags & STRM_UTF8)
    {
      int tmp = utf8min (this->rdbuf.curr, nbytes);
      bytes -= nbytes - tmp;
      nbytes = tmp;
    }

  memcpy (dst, this->rdbuf.curr, nbytes);
  this->rdbuf.bump (nbytes);
  this->add_pos (bytes);
  return ((int)bytes);
}

int stream::write (interpreter *interp, const void *src, uint32_t bytes)
{
  int nbytes = (int)bytes;

  if (this->need_rflush () && !this->rflush (interp))
    return (-1);

  this->set_lastw ();
  int cnt = min (this->wrbuf.left (), nbytes);

  /* For UTF-8 streams, we have to make sure that we don't
   * write an incomplete character. */
  if (this->io_flags & STRM_UTF8)
    cnt = utf8min ((const char *)src, cnt);
  
  memcpy (this->wrbuf.curr, src, cnt);
  this->wrbuf.bump (cnt);
  src = (const char *)src + cnt;

  if ((nbytes -= cnt) > 0)
    { // There's still data to write.
      if (!this->write_full (interp, this->wrbuf.start, this->wrbuf.used ()))
        return (-1);
      else if (nbytes > this->bufsize ())
        { // Write directly to handle.
          int tmp = nbytes - (nbytes % this->bufsize ());

          if (this->io_flags & STRM_UTF8)
            for (; (((const unsigned char *)src)[tmp] & 0xc0) == 0x80; ++tmp) ;

          if (!this->write_full (interp, src, tmp))
            return (-1);
          
          src = (const char *)src + tmp, nbytes -= tmp;
        }

      memcpy (this->wrbuf.start, src, nbytes);
      this->wrbuf.set (this->wrbuf.start + nbytes);
    }

  this->add_pos (bytes);
  return ((int)bytes);
}

int stream::putb (interpreter *interp, int byte)
{
  if ((this->need_rflush () && !this->rflush (interp)) ||
      (this->wrbuf.left () == 0 && !this->wflush (interp)))
    return (-1);

  this->set_lastw ();
  *this->wrbuf.curr++ = (unsigned char)byte;
  this->add_pos (1);
  return (1);
}

int stream::nputb (interpreter *interp, unsigned int n, int byte)
{
  if (n == 0)
    return (0);
  else if (this->need_rflush () && !this->rflush (interp))
    return (-1);

  int off = n, cnt = min (this->wrbuf.left (), off);
  uint32_t bsz = this->bufsize ();

  memset (this->wrbuf.curr, byte, cnt);
  this->wrbuf.bump (cnt);
  this->set_lastw ();

  if ((n -= cnt) > 0)
    {
      if (!this->write_full (interp, this->wrbuf.start, bsz))
        return (-1);
      
      for (cnt = n / bsz; cnt != 0; --cnt)
        {
          char *p = (char *)memset (this->wrbuf.start, byte, bsz);
          if (!this->write_full (interp, p, bsz))
            return (-1);
        }
      
      cnt = n % bsz;
      memset (this->wrbuf.start, byte, cnt);
      this->wrbuf.set (this->wrbuf.start + cnt);
    }

  this->add_pos (off);
  return (off);
}

int stream::putuc (interpreter *interp, uint32_t ch)
{
  unsigned char buf[4];
  int i = 0, len = u32tou8 (buf, ch);

  if ((this->need_rflush () && !this->rflush (interp)) ||
      (this->wrbuf.left () == 0 && !this->wflush (interp)))
    return (-1);

  this->set_lastw ();
  
  do
    *this->wrbuf.curr++ = buf[i++];
  while (i < len);

  this->add_pos (i);
  return (i);
}

int stream::wbase (interpreter *interp, int base)
{
  char s[4];   // Enough for "36r
  char *p = s;
  
  switch (base)
    {
      case 10:
        return (0);
      case 8:
        *p++ = '0';
        break;
      case 16:
        p[0] = '0', p[1] = 'x';
        p += 2;
        break;
      case 2:
        p[0] = '0', p[1] = 'b';
        p += 2;
        break;
      default:
        {
          int c1 = base / 10;
          int c2 = base % 10;

          if (c1 != 0)
            *p++ = (char)(c1 + '0');
          *p++ = (char)(c2 + '0');
          *p++ = 'r';
          break;
        }
    }

  return (this->write (interp, s, (int)(p - s)));
}

int stream::peekb (interpreter *interp)
{
  if ((this->need_wflush () && !this->wflush (interp)) ||
      (this->rdbuf.left () == 0 && !this->refill (interp)))
    return (-1);

  this->set_lastr ();
  return ((unsigned char)*this->rdbuf.curr);
}

int stream::getb (interpreter *interp)
{
  int ret = this->peekb (interp);
  if (ret >= 0)
    {
      this->rdbuf.bump (1);
      this->add_pos (1);
    }

  return (ret);
}

static bool strm_sgetc (interpreter *interp, stream& strm, schar& ch)
{
  if ((strm.need_wflush () && !strm.wflush (interp)) ||
      (strm.rdbuf.left () == 0 && !strm.refill (interp)))
    return (false);

  strm.set_lastr ();
  ch.buf[(ch.len = 1) - 1] = (unsigned char)*strm.rdbuf.curr++;
  int i = UTF8_SKIP[(uint8_t)ch.buf[0]];

  if (i == 0)
    return (false);
  for (int j = 1; j < i; ++j)
    {
      if (strm.rdbuf.left () == 0 && !strm.refill (interp))
        return (false);

      ch.buf[ch.len++] = (uint8_t)*strm.rdbuf.curr++;
    }

  ch.uc = u8tou32 (ch.buf, ch.len);
  strm.add_pos (ch.len);
  return (true);
}

bool stream::sgetc (interpreter *interp, schar& ch)
{
  bool ret = strm_sgetc (interp, *this, ch);
  if (!ret)
    ch.uc = UEOF;

  return (ret);
}

uint32_t stream::getuc (interpreter *interp)
{
  schar ch;
  return (!strm_sgetc (interp, *this, ch) ? UEOF : ch.uc);
}

bool stream::seek (interpreter *interp, object off, int whence)
{
  bool ret = false;

  if (this->ops->seek && (whence == SEEK_SET ||
      whence == SEEK_CUR || whence == SEEK_END))
    {
      if (this->need_wflush () && !this->wflush (interp))
        return (ret);

      ret = this->xseek (interp, spos::decode (off), whence);
      if (ret)
        {
          this->reset_buff ();
          this->io_flags &= ~(STRM_EOS | STRM_ERR);
        }
    }

  return (ret);
}

bool stream::ungetuc (const unsigned char *p, int n)
{
  if (!(this->io_flags & (STRM_LASTR | STRM_BID)) ||
      n <= 0 || this->rdbuf.used () < n)
    return (false);

  this->add_pos (-n);

  do
    *--this->rdbuf.curr = p[--n];
  while (n != 0);

  return (true);
}

void stream::lock (interpreter *interp)
{
#ifndef KP_NO_THREADS
  if (!(this->io_flags & STRM_NOLOCK))
    as_lock(this->ilock)->grab (interp, UNBOUND, false);
#endif
}

bool stream::trylock (interpreter *interp)
{
#ifndef KP_NO_THREADS
  return ((this->io_flags & STRM_NOLOCK) == 0 ||
    as_lock(this->ilock)->trygrab (interp) == 0);
#endif
}

void stream::unlock (interpreter *interp)
{
#ifndef KP_NO_THREADS
  if (!(this->io_flags & STRM_NOLOCK))
    as_lock(this->ilock)->drop (interp);
#endif
}

bool stream::close (interpreter *interp)
{
  bool ret = true;
  
  if (this->io_flags & STRM_CLOSED)
    ;   // Nothing to do.
  else if (this->wrbuf.used () > 0 && !this->wflush (interp))
    return (false);
  else if (this->ops->close &&
      (ret = this->ops->close (interp, *this)))
	this->io_flags |= STRM_CLOSED;

  return (ret);
}

static inline
fhandle_t as_osfh (void *cookie)
{
  return ((fhandle_t)(intptr_t)cookie);
}

static inline int
osfh_readb (interpreter *interp, stream& strm, void *dst, uint32_t nb)
{
  return (llio_readb (interp, as_osfh (strm.cookie), (char *)dst, nb));
}

static inline int
osfh_writeb (interpreter *interp, stream& strm, const void *ptr, uint32_t nb)
{
  return (llio_writeb (interp, as_osfh (strm.cookie), (const char *)ptr, nb));
}

static inline bool
osfh_seekb (interpreter *interp, stream& strm, spos& pos, int whence)
{
  return (llio_seek (interp, as_osfh (strm.cookie), pos.offset, whence));
}

static inline bool
osfh_close (interpreter *, stream& strm)
{
  return (llio_close (as_osfh (strm.cookie)));
}

static const stream::xops bin_ops =
{
  osfh_readb,
  osfh_writeb,
  osfh_seekb,
  osfh_close
};

#ifdef KP_PLATFORM_WINDOWS

static inline int
osfh_readt (interpreter *interp, stream& strm, void *dst, uint32_t nb)
{
  return (llio_readt (interp, as_osfh (strm.cookie), (char *)dst, nb));
}

static inline int
osfh_writet (interpreter *interp, stream& strm, const void *ptr, uint32_t nb)
{
  return (llio_writet (interp, as_osfh (strm.cookie), (const char *)ptr, nb));
}

static inline bool
osfh_seekt (interpreter *interp, stream& strm, spos& pos, int whence)
{
  if (pos.offset != 0)
    return (false);

  return (osfh_seekb (interp, strm, pos, whence));
}

static const stream::xops text_ops =
{
  osfh_readt,
  osfh_writet,
  osfh_seekt,
  osfh_close
};

static inline const stream::xops*
fhandle_ops (uint32_t flags)
{
  return ((flags & STRM_BIN) ? &bin_ops : &text_ops);
}

#else

static inline const stream::xops*
fhandle_ops (uint32_t)
{
  return (&bin_ops);
}

#endif

stream* fstream_open (interpreter *interp,
  const char *path, const char *mode)
{
  int flags = 0;
  fhandle_t fh = llio_open (interp, path, mode, &flags);

  if (!fhandle_valid_p (fh))
    return (nullptr);

  const stream::xops *ops = fhandle_ops (flags);
  try
    {
      stream *ret = stream::make (interp, flags,
        STRM_BUFSIZ, ops, (void *)(intptr_t)fh);
      if (ret->io_flags & STRM_WRITE)
        // Need to flush this object upon exit.
        ret->vo_full |= FLAGS_ATEXIT;

      return (ret);
    }
  catch (...)
    {
      llio_close (fh);
      throw;
    }
}

bool fstream_fstat (interpreter *interp, stream *strm, fhandle_stat& out)
{
  return (llio_fhstat (interp, as_osfh (strm->cookie), out));
}

bool fstream_truncate (interpreter *interp, stream *strm, uint64_t nsize)
{
  return (strm->flush (interp) &&
    llio_truncate (as_osfh (strm->cookie), nsize));
}

// Standard streams initialisation.

struct std_stream
{
  bvector bvec;
  char buffer[STRM_BUFSIZ];
  stream strm;
};

static stream*
init_std_stream (interpreter *interp, std_stream& s,
                 uint32_t flags, fhandle_t fh)
{
  stream& strm = s.strm;

  strm.rdbuf.init (s.buffer, sizeof (s.buffer));
  strm.rdbuf.bump (sizeof (s.buffer));
  strm.wrbuf.init (s.buffer, sizeof (s.buffer));

  s.bvec.data = (unsigned char *)s.buffer;
  s.bvec.nbytes = sizeof (s.buffer);
  s.bvec.vo_type = typecode::BVECTOR;
  strm.bvec = s.bvec.as_obj ();

  strm.ops = fhandle_ops (STRM_BIN);
  strm.ilock = alloc_lock (interp, true);

  strm.vo_type = typecode::STREAM;
  strm.pos = spos (0);
  strm.io_flags = STRM_UTF8 | STRM_BUILTIN | flags;
  strm.cookie = (void *)(intptr_t)fh;
  strm.extra = UNBOUND;

  return (&strm);
}

// External definitions.
object in_stream;
object out_stream;
object err_stream;

static int
do_init_streams (interpreter *interp)
{
  int ret = init_op::call_deps (interp, &init_symbols, &init_llio);
  if (ret != init_op::result_ok)
    return (ret);

  static std_stream std_streams[3];
  stream *p;

  p = init_std_stream (interp, std_streams[0], STRM_READ, llio_stdfh (0));
  in_stream = ensure_mask(p)->as_obj ();

  p = init_std_stream (interp, std_streams[1], STRM_WRITE, llio_stdfh (1));
  out_stream = ensure_mask(p)->as_obj ();

  p = init_std_stream (interp, std_streams[2], STRM_WRITE, llio_stdfh (2));
  err_stream = ensure_mask(p)->as_obj ();

  // Add the dynamic bindings.
  object sym = intern (interp, "*in*", 4, nullptr, symbol::special_flag);
  symval(sym) = in_stream;

  sym = intern (interp, "*out*", 5, nullptr, symbol::special_flag);
  symval(sym) = out_stream;

  sym = intern (interp, "*err*", 5, nullptr, symbol::special_flag);
  symval(sym) = err_stream;

  return (init_op::result_ok);
}

init_op init_streams (do_init_streams, "streams");

// XXX: Better definition.
const object EOS = fixint (0) | EXTRA_BIT;

KP_DECLS_END
